package cn.snow.iot.module.cps.service.equipment;

import cn.hutool.core.bean.BeanUtil;
import cn.snow.iot.framework.common.pojo.CommonResult;
import cn.snow.iot.module.cps.api.equipment.dto.EquipmentRelateWarningDTO;
import cn.snow.iot.module.cps.constant.CpsConstant;
import cn.snow.iot.module.cps.controller.admin.equipment.vo.EquipmentPage;
import cn.snow.iot.module.cps.controller.admin.equipment.vo.EquipmentPageVO;
import cn.snow.iot.module.cps.controller.admin.equipment.vo.SearchParamVO;
import cn.snow.iot.module.cps.controller.admin.home.vo.EquipmentAgreementTypeVO;
import cn.snow.iot.module.cps.controller.admin.home.vo.EquipmentStatusVO;
import cn.snow.iot.module.cps.dal.dataobject.equipment.EquipmentDO;
import cn.snow.iot.module.cps.dal.dataobject.equipment.EquipmentPointDO;
import cn.snow.iot.module.cps.dal.mysql.equipment.EquipmentMapper;
import cn.snow.iot.module.cps.ekuiperClient.MonitorClient;
import cn.snow.iot.module.cps.utils.DateUtil;
import cn.snow.iot.module.warning.enums.api.config.AdminConfigApi;
import cn.snow.iot.module.warning.enums.api.config.vo.ConfigCountVO;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xxdb.data.BasicTable;
import com.xxdb.data.DolphinDBUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@Service
@Slf4j
public class EquipmentServiceImpl extends ServiceImpl<EquipmentMapper, EquipmentDO> implements EquipmentService {

    @Resource
    private EquipmentPointService pointService;
    @Autowired
    @Lazy
    private DolphinDBUtils dolphindbUtils;

    @Autowired
    private MonitorClient monitorClient;

    @Resource
    private AdminConfigApi configApi;

    @Value("${ekuiper.cps.url}")
    private String ekuiperUrl;

    @Value("${ekuiper.cps.ws}")
    private String ws;

    @Value("${ekuiper.cps.sql}")
    private String ekuiperSql;

    @Value("${ekuiper.cps.actions.path}")
    private String ekuiperWebSocketPath;

    @Resource
    private DateUtil dateUtil;

    @Override
    public IPage<EquipmentPage> selectEquipmentPage(EquipmentPageVO equipment) {
        IPage<EquipmentPage> page = new Page<>(equipment.getPageNo(), equipment.getPageSize());
        IPage<EquipmentPage> equipmentDOIPage = baseMapper.selectEquipmentPage(page, equipment);
        List<EquipmentPage> records = equipmentDOIPage.getRecords();
        List<Long> ids = new ArrayList<>();
        records.forEach(item -> {
             ids.add(item.getId());
        });
        if (ids.size() > 0) {
            List<ConfigCountVO> warningCount = configApi.getWarningCount(ids);
            // 创建一个Map来快速查找ConfigCountVO对象
            Map<Long, Integer> configCountMap = new HashMap<>();
            for (ConfigCountVO configCount : warningCount) {
                configCountMap.put(configCount.getEquipmentId(), configCount.getWarningCount());
            }
            // 遍历EquipmentDO列表，并更新warningCount
            for (EquipmentPage equipmentPage : records) {
                Integer count = configCountMap.get(equipmentPage.getId());
                if (count != null) {
                    equipmentPage.setWarningCount(count);
                } else {
                    equipmentPage.setWarningCount(0);
                }
            }
        }
        page.setRecords(records);
        return equipmentDOIPage;
    }

    @Override
    public Boolean enableEquipment(List<Long> ids) {
        for (Long id : ids) {
            EquipmentDO equipment = this.lambdaQuery().eq(EquipmentDO::getId, id).one();
            if (BeanUtil.isEmpty(equipment)) {
                return false;
            }
            //调用dolphinDB脚本去新建该设备对应的数据表
            List<String> pointCodes = new ArrayList<>();
            createDolphinTable(equipment, pointCodes);
            // 启用,前端系统字典的is_enable的1代表启用
            equipment.setIsEnable(1);
            updateById(equipment);
        }
        return true;
    }

    @Override
    public Boolean disenableEquipment(List<Long> ids) {
        return lambdaUpdate().in(EquipmentDO::getId, ids).set(EquipmentDO::getIsEnable, 0).update();
    }

    @Override
    public Boolean removeEquipment(EquipmentPage equipment) {
        Long id = equipment.getId();
        if (equipment.getWarningCount() > 0) {
            // 调用告警服务
            boolean status = configApi.deleteByEquipmentId(equipment.getId());
            if (status) {
                pointService.removePoint(id);
                baseMapper.deleteById(id);
                return true;
            } else {
                return false;
            }
        } else {
            pointService.removePoint(id);
            baseMapper.deleteById(id);
            return true;
        }
    }

    @Override
    public Object restart() {
        // 查询所有已启用的设备信息
        List<EquipmentDO> lists = lambdaQuery().eq(EquipmentDO::getIsEnable, 1).list();
        if (lists.isEmpty()) {
            return false; // 如果没有已启用的设备，直接返回
        }
        List<Long> ids = lists.stream().map(EquipmentDO::getId).collect(Collectors.toList());

        // 将查询的设备全部失效
        LambdaUpdateWrapper<EquipmentDO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
        lambdaUpdateWrapper.set(EquipmentDO::getIsEnable, 0).in(EquipmentDO::getId, ids);
        boolean updateSuccess = this.update(lambdaUpdateWrapper);

        if (updateSuccess) {
            // 将查询的设备全部启用
            enableEquipment(ids);
        } else {
            // 处理更新失败的情况，例如记录日志或者抛出异常
            log.error("Failed to disable equipment");
            return false;
        }
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public CommonResult submit(EquipmentDO equipment) {
        List<EquipmentDO> equipmentList = checkCode(equipment);
        if (equipmentList.size() > 0) {
//            throw new RuntimeException("设备编码已存在: " + equipment.getEquipmentCode());
            return CommonResult.error(207, "设备编码已存在");
        }
        if (equipment.getId() == null) {
            equipment.setEquipmentAscription("none");
        }
        if (equipment.getAgreementType() == null) {
            equipment.setAgreementType("");
        }
        saveOrUpdate(equipment);
        return CommonResult.success(equipment);
    }

    @Override
    public Boolean relevancyEquipment(List<Long> ids, Long passageId) {
        return baseMapper.relevancyEquipment(ids, passageId);
    }

    @Override
    public Boolean deleteEquipmentByPassageId(Long id) {
        baseMapper.deleteEquipmentByPassageId(String.valueOf(id));
        return true;
    }

    /**
     * 当数据监控界面中选择某个设备后，调用此逻辑实现单个设备的点位数据监控
     * 此业务包含两部分 1.查询设备信息 2.根据设备创建ekuiper规则，借助ekuiper的websocket动作进行实时数据推送前端
     * @param id
     * @return
     */
    @Override
    public EquipmentDO detailMonitor(Long id) {
        //根据设备id获取设备基础信息
        EquipmentDO equipmentDO = getById(id);
        //根据设备编码调用ekuiper接口，生成一个输出动作为websocket的规则，规则名称取 设备编码+_Monitor  动作作为websocket服务端，供前端数采监控连接
        JSONObject actionParam = new JSONObject();
        actionParam.put("path", ekuiperWebSocketPath);
        actionParam.put("insecureSkipVerify", true);
        actionParam.put("sendSingle", true);

        JSONObject action = new JSONObject();
        action.put("websocket", actionParam);

        JSONArray actions = new JSONArray();
        actions.add(action);

        JSONObject data = new JSONObject();
        //规则id，唯一   monitor-设备编码
        data.put("id", "monitor-" + equipmentDO.getEquipmentCode());
        //规则名称，数据监控-设备名称
        data.put("name", "数据监控-" + equipmentDO.getEquipmentName());
        data.put("sql", ekuiperSql.replace("param", equipmentDO.getEquipmentCode()));
        data.put("actions", actions);
        monitorClient.ruleCreation(data, ekuiperUrl);
        equipmentDO.setWebSocket(ws + ekuiperWebSocketPath);
        return equipmentDO;
    }

    @Override
    public IPage<EquipmentPage> selectEquipmentPageByProductCode(EquipmentPageVO equipment) {
        IPage<EquipmentPage> page = new Page<>(equipment.getPageNo(), equipment.getPageSize());
        IPage<EquipmentPage> equipmentDOIPage = baseMapper.selectEquipmentPage(page, equipment);
        List<EquipmentPage> records = equipmentDOIPage.getRecords();
        List<Long> ids = new ArrayList<>();
        records.forEach(item -> {
            ids.add(item.getId());
        });
        if (ids.size() > 0) {
            List<ConfigCountVO> warningCount = configApi.getWarningCount(ids);
            // 创建一个Map来快速查找ConfigCountVO对象
            Map<Long, Integer> configCountMap = new HashMap<>();
            for (ConfigCountVO configCount : warningCount) {
                configCountMap.put(configCount.getEquipmentId(), configCount.getWarningCount());
            }
            // 遍历EquipmentDO列表，并更新warningCount
            for (EquipmentPage equipmentPage : records) {
                Integer count = configCountMap.get(equipmentPage.getId());
                if (count != null) {
                    equipmentPage.setWarningCount(count);
                } else {
                    equipmentPage.setWarningCount(0);
                }
            }
        }
        page.setRecords(records);
        return equipmentDOIPage;
    }

    private List<EquipmentDO> checkCode(EquipmentDO equipment) {
        LambdaQueryWrapper<EquipmentDO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        if (equipment.getEquipmentCode().isEmpty()) {
            throw new RuntimeException("设备编码不能为空");
        }
        lambdaQueryWrapper.eq(EquipmentDO::getEquipmentCode, equipment.getEquipmentCode());
        //id不为空 是修改 排除当前数据
        if (equipment.getId() != null) {
            lambdaQueryWrapper.ne(EquipmentDO::getId, equipment.getId());
        }
        return baseMapper.selectList(lambdaQueryWrapper);
    }

    /**
     * 根据设备的模型信息构造脚本需要的参数并执行
     *
     * @param equipment  equipment
     * @param pointCodes pointCodes
     * @return boolean
     */
    public boolean createDolphinTable(EquipmentDO equipment, List<String> pointCodes) {
        JSONArray jsonArray = new JSONArray();
        Map<String, Integer> type = new HashMap<>();
        type.put("bool", 1);
        type.put("float", 16);
        type.put("string", 18);

        List<EquipmentPointDO> equipmentPoints = pointService.lambdaQuery().eq(EquipmentPointDO::getEquipmentId, equipment.getId()).list();


        //获取需要原始订阅的测点编码
        pointCodes.addAll(equipmentPoints.stream().filter(e -> {
            return CpsConstant.ISPUSH_STATUS.equals(e.getIsEnable());
        }).map(EquipmentPointDO::getPointCode).collect(Collectors.toList()));


        equipmentPoints.forEach(e -> {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("name", e.getPointCode());
            jsonObject.put("type", type.get(e.getDataType()));
            jsonArray.add(jsonObject);
        });

        boolean dolph = dolphindbUtils.createDeviceTable(equipment, jsonArray);
        log.info("dolphinDB建表执行结果: {}", dolph);
        return dolph;
//        return true;
    }

    @Override
    public List<EquipmentAgreementTypeVO> getAgreementTypeCount() {
        return baseMapper.getAgreementTypeCount();
    }

    @Override
    public List<EquipmentRelateWarningDTO> getEquipmentName(List<Long> equipmentIds) {
        return baseMapper.getEquipmentName(equipmentIds);
    }

    @Override
    public List<EquipmentStatusVO> getIsEnableCount() {
        return baseMapper.getIsEnableCount();
    }

    @Override
    public CommonResult cancelPassage(String id) {
        boolean status = baseMapper.cancelPassage(id);
        return CommonResult.success(status);
    }

    /**
     * 今日 today，本周 Week，本月 month，开始到结束 dataChange
     *
     * @param searchParamVO searchParamVO 开始时间结束时间
     * @param distinction   distinction 状态
     * @return 时序数据库查询结果
     */
    @Override
    public CommonResult cardDate(SearchParamVO searchParamVO, String distinction) {
        return CommonResult.success(cardHistory(searchParamVO, distinction));
    }


    /**
     * cardDate调用  exportCard 导出历史数据调用
     *
     * @param searchParamVO searchParamVO 开始时间结束时间
     * @param distinction   distinction 状态
     * @return 时序数据库查询结果
     */
    public Map<String, List> cardHistory(SearchParamVO searchParamVO, String distinction) {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss");
        switch (distinction) {
            case CpsConstant.TODAY:
                //今天
                searchParamVO.setBeginDate(transferString2Date(dateUtil.getTodayStartTime()));
                searchParamVO.setEndDate(transferString2Date(dateUtil.getTodayEndTime()));
                break;
            case CpsConstant.WEEK:
                //本周
                searchParamVO.setBeginDate(transferString2Date(dateUtil.getBeginDayOfWeek().toString()));
                searchParamVO.setEndDate(transferString2Date(dateUtil.getEndDayOfWeek().toString()));
                break;
            case CpsConstant.MONTH:
                //本月
                searchParamVO.setBeginDate(transferString2Date(dateUtil.getMonthTime().get("startDate").toString()));
                searchParamVO.setEndDate(transferString2Date(dateUtil.getMonthTime().get("endDate").toString()));
                break;
            default:
        }
        // 执行时序数据库脚本进行查询测定
        BasicTable data = dolphindbUtils.searchHistoryData(searchParamVO.getDeviceCode(), searchParamVO.getPointCode(), sf.format(searchParamVO.getBeginDate()), sf.format(searchParamVO.getEndDate()));
        if (StringUtils.isEmpty(data)) {
            return null;
        }
        Map<String, List> result = new HashMap();
        List<Date> xdata = new ArrayList<>();
        List<String> ydata = new ArrayList<>();
        String columnName = searchParamVO.getPointCode();
        int rows = data.rows();
        IntStream.range(0, rows).forEach(index -> {
            String rowJson = data.getRowJson(index);
            JSONObject o = JSON.parseObject(rowJson);
            try {
                SimpleDateFormat format = new SimpleDateFormat("yyyy.MM.dd'T'HH:mm:ss.SSS");
                Date thetime = format.parse(o.getString("thetime"));
                xdata.add(thetime);
                ydata.add(o.getString(columnName));
            } catch (ParseException e) {
                log.warn("历史数据查询，构造数据结构失败");
                throw new RuntimeException(e);
            }
        });
        result.put("xdata", xdata);
        result.put("ydata", ydata);
        return result;
    }

    /**
     * String转换为Date类型
     *
     * @param s 传入的类型
     * @return 返回date
     */
    public Date transferString2Date(String s) {
        Date date = new Date();
        try {
            date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(s);
        } catch (ParseException e) {
            e.toString();
        }
        return date;
    }
}
